use crossbeam_deque::Injector;
use fudata::db;
use fudata::model::kline::KlineList;
use fudata::model::trade::trade_holding::TradeHoldingList;
use fudata::model::trade::trade_order::TradeOrderList;
use fudata::model::trade::trade_strategy::TradeStrategy;
use fudata::model::trade::trade_strategy::TradeStrategyList;
use log::error;
use log::trace;
use std::sync::Arc;
use anyhow::anyhow;
use anyhow::Result;
use log::info;
use zenoh::key_expr::KeyExpr;
use zenoh::prelude::r#async::*;
use zenoh::queryable::Query;

use crate::qtrade::TRADE_HOLDINGS_TOPIC;
use crate::qtrade::TRADE_ORDERS_TOPIC;
use crate::qtrade::TRADE_STRATEGIES_TOPIC;

use super::strategy::TRADE_STRATEGIES;
use super::RECEIVE_TIMEOUT_MS;

// Create a worker pool using Tokio
pub fn start_worker_pool(num_workers: u8) -> Arc<Injector<Query>> {
    let injector = Arc::new(Injector::new());

    // Spawn workers directly using tokio::spawn
    for worker_no in 0..num_workers {
        let ij_clone = Arc::clone(&injector);
        tokio::spawn(async move {
            loop {
                let task = ij_clone.steal().success();

                match task {
                    Some(trade_query) => {
                        // Process the trade_query using the handle_query function
                        let _sample = handle_query(worker_no, &trade_query).await;
                        // Do something with the sample...
                    }
                    None => {
                        trace!("(worker_query) Workder[{worker_no}] steal no trade_query, wait {}ms", RECEIVE_TIMEOUT_MS);
                        tokio::time::sleep(
                            std::time::Duration::from_millis(RECEIVE_TIMEOUT_MS)
                        ).await;
                    }
                }
            }
        });
    }

    injector
}

// Define the handle_query function
async fn handle_query(worker_no: u8, query: &Query) -> Result<()> {
    // Your actual query processing logic goes here
    // ...

    let selector = query.selector();
    info!(
        "\n\n (handle_query) Worker[{worker_no}] Received query with selector key_expr={:?} parameters={} selector.parameters_stringmap()={:?}\n\n",
        selector.key_expr,
        selector.parameters(),
        selector.parameters_stringmap()
    );
    let key = selector.key_expr.as_str();
    match key {
        TRADE_STRATEGIES_TOPIC => {
            handle_query_trade_strategies(&query).await?;
        }
        TRADE_ORDERS_TOPIC => {
            handle_query_trade_orders(&query).await?;
        }
        TRADE_HOLDINGS_TOPIC => {
            handle_query_trade_holdings(&query).await?;
        }
        _ => {
            error!("unknown key_expr:{}", key);
        }
    }

    Ok(())
}

async fn handle_query_trade_strategies(query: &Query) -> Result<()> {
    let key_expr = query.key_expr().clone();

    let strategies: Vec<TradeStrategy> = TRADE_STRATEGIES.to_vec();
    let ss: zenoh::value::Value = TradeStrategyList::from(strategies).into();

    let sample = Sample::new(key_expr, ss);
    let rs = query.reply(Ok(sample)).res().await;

    match rs {
        Ok(_) => Ok(()),
        Err(e) => Err(anyhow!(e)),
    }
}

async fn handle_query_trade_orders(query: &Query) -> Result<()> {
    let key_expr = query.key_expr().clone();

    let selector = query.selector();

    let mut last_updated: Option<String> = None;
    // Extract parameters from the selector
    match selector.parameters_stringmap() {
        Ok(params) => {
            if let Some(value) = params.get("_time") {
                // Note: Using "_time" as placeholder; correct key should be used
                info!("Found parameter '_time': {}", value);
            }
            if let Some(ts) = params.get("last_updated") {
                info!("Last updated parameter: {}", ts);
                last_updated = Some(ts.to_string());
            }
        }
        Err(err) => {
            error!("Failed to parse parameters: {}", err);
        }
    }
    let trade_orders = db::trade::trade_order::select(None, None, None, 0, 10).await?;
    let ss: zenoh::value::Value = TradeOrderList::from(trade_orders).into();
    let sample = Sample::new(key_expr, ss);
    let rs = query.reply(Ok(sample)).res().await;

    match rs {
        Ok(_) => Ok(()),
        Err(e) => Err(anyhow!(e)),
    }
}

async fn handle_query_trade_holdings(query: &Query) -> Result<()> {
    let key_expr = query.key_expr().clone();

    let selector = query.selector();
    let mut trade_holding_id: Option<String> = None;
    let mut last_updated: Option<String> = None;
    // Extract parameters from the selector
    match selector.parameters_stringmap() {
        Ok(params) => {
            if let Some(value) = params.get("trade_holding_id") {
                // Note: Using "_time" as placeholder; correct key should be used
                info!("Found parameter 'trade_holding_id': {}", value);
                trade_holding_id = Some(value.to_string());
            }
            if let Some(ts) = params.get("last_updated") {
                info!("Last updated parameter: {}", ts);
                last_updated = Some(ts.to_string());
            }
        }
        Err(err) => {
            error!("Failed to parse parameters: {}", err);
        }
    }
    
    let holdings = db::trade::trade_holding::select(trade_holding_id, 0, 10).await?;
    let ss: zenoh::value::Value = TradeHoldingList::from(holdings).into();

    let sample = Sample::new(key_expr, ss);
    let rs = query.reply(Ok(sample)).res().await;

    match rs {
        Ok(_) => Ok(()),
        Err(e) => Err(anyhow!(e)),
    }
}
